/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.Vector;

import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.MergeView;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
import org.jgroups.logging.Log;
import org.jgroups.logging.LogFactory;
import org.jgroups.util.Util;
import org.sabayframework.cluster.ClusterEvent;
import org.w3c.dom.Element;


/**
 * Class responsible for cluster membership and cluster-wide communications.
 * @author Alex Antonau
 * */
final class ClusterConnection implements Receiver {

	private static final long STATE_TRANSFER_TIMEOUT = 30000l;
	private JChannel channel;
    private final Vector<Address> members=new Vector<Address>();
    private Address localAddr=null;
    private String clusterName;
    private Element props;
	private static final Log log = LogFactory.getLog(ClusterConnection.class);
	private ClusterState clusterState;
	
	ClusterConnection(String clusterName, Element props, ClusterState clusterState) {
		this.clusterName = clusterName;
		this.props = props;
		this.clusterState = clusterState;
	}
	
	public void start()throws Exception 
	{
		if(channel==null){
			channel = new JChannel(props);
			channel.setDiscardOwnMessages(true);
			channel.setReceiver(this);
			channel.connect(clusterName);
			channel.getState(null, STATE_TRANSFER_TIMEOUT);
		}
	}
	
    public void stop() {
        if(channel != null) {
            channel.close();  // disconnects from channel and closes it
            channel=null;
        }
    }
    public Address getLocalAddress() {
        if(localAddr != null) return localAddr;
        if(channel != null)
            localAddr=channel.getAddress();
        return localAddr;
    }
    
    public void sendEvent(ClusterEvent event) throws Exception {
    	Message msg=new Message();
        final ByteArrayOutputStream out_stream=new ByteArrayOutputStream(512);
        DataOutputStream out=new DataOutputStream(out_stream);
        event.writeTo(out);
    	byte[] data=out_stream.toByteArray();
    	msg.setBuffer(data);
    	channel.send(msg);
    }
    
    public boolean isCoordinator() {
    	return channel == null? false : Util.isCoordinator(channel);
    }
    
    
	@Override
	public void getState(OutputStream out) throws Exception {
		clusterState.getState(out);
	}

	@Override
	public void setState(InputStream in) throws Exception {
		clusterState.mergeState(in);
	}

	@Override
	public void unblock() {
	}

	@Override
	public void receive(Message msg) {
        try {
    		byte[] data=msg.getBuffer();
            ByteArrayInputStream in_stream=new ByteArrayInputStream(data);
            DataInputStream in=new DataInputStream(in_stream);
            ClusterEvent event = new ClusterEvent();
			event.readFrom(in);
			clusterState.handleEvent(event);
		} catch (Exception e) {
			log.error(msg.toString(), e);
		}
	}
	@Override
	public void block() {
	}
	@Override
	public void suspect(Address suspect) {
	}
	@Override
	public void viewAccepted(View view) {
        if(view == null) return;
        List<Address> joined_mbrs, left_mbrs, tmp;
        tmp=view.getMembers();

        synchronized(members) {
            // get new members
            joined_mbrs= Util.newMembers(members, tmp);
            // get members that left
            left_mbrs=Util.leftMembers(members, tmp);
            // adjust our own membership
            members.removeAllElements();
            members.addAll(tmp);
        }
        if(joined_mbrs!=null && joined_mbrs.size() > 0) clusterState.membersJoined(joined_mbrs);
        if(left_mbrs!=null && left_mbrs.size() > 0) clusterState.membersLeft(left_mbrs);
        if(view instanceof MergeView) {
        	// merge state
            if(!joined_mbrs.isEmpty()) {
                Address coord=joined_mbrs.get(0);
                Address local_addr=channel.getAddress();
                if(local_addr != null && !local_addr.equals(coord)) {
                    try {
                        channel.getState(coord, STATE_TRANSFER_TIMEOUT);
                    }
                    catch(Exception e) {
                        log.error("merge state from"+coord, e);
                    }
                }
            }
        }
        
	}

}
